package edu.hawaii.ics.pjdb.query_planner;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.hawaii.ics.pjdb.result_transport.TupleListener;
import edu.hawaii.ics.pjdb.result_transport.TupleServiceHelper;
import edu.hawaii.ics.pjdb.types.Plan;
import edu.hawaii.ics.pjdb.types.WorkerCommand;
import edu.hawaii.ics.pjdb.util.ThriftHelper;
import edu.hawaii.ics.pjdb.worker.WorkerHelper;

public class PlanRunner {
    protected static final Logger logger = LoggerFactory
            .getLogger(PlanRunner.class);

    /**
     * All commands that get passed to workers will run inside this thread pool.
     * Change it to control how many simultaneous commands can be run.
     * 
     */
    protected ExecutorService commandPool = Executors.newCachedThreadPool();

    /**
     * Runs a query plan. Returns a semaphore with an initially non-positive
     * number of permits so that the caller can block until the plan is finished
     * running if need be. Usage:
     * 
     * Semaphore lock = planRunner.runPlan(commands); lock.acquire(); // blocks
     * until 'commands' are done running.
     * 
     * Results will be sent to the {@link TupleListener} outputLocation.
     * 
     * @param commands
     * @param outputLocation
     * @throws TException
     * @return {@link Semaphore}
     */
    public Semaphore runPlan(Plan commands, TupleListener outputLocation)
            throws TException {
        Semaphore lock = new Semaphore(-commands.getCommandsSize() + 1);

        for (WorkerCommand command : commands.getCommands()) {
            // Start an output listener if specified. Note that we do not start
            // an output
            // listener if remote is set to true because that means a worker
            // should be
            // doing it.
            if (command.isSetOutputAddress()
                    && !command.getOutputAddress().isRemote()) {
                startOutputListener(command, outputLocation);
            }

            logger.info("Sending command to worker with address: {}",
                    command.getWorkerAddress());
            sendCommand(command, lock);
        }

        return lock;
    }

    /**
     * Internal helper method that starts an output listener in a separate
     * thread.
     * 
     * @param command
     * @param listener
     * @throws TTransportException
     */
    protected void startOutputListener(WorkerCommand command,
            final TupleListener listener) throws TTransportException {
        final int port = command.getOutputAddress().getPort();
        logger.info("Starting tuple server on port {}", port);

        ThriftHelper.startServer(TupleServiceHelper
                .createServer(port, listener));
    }

    /**
     * Runs the provided command in a separate thread. A permit from the
     * provided {@link Semaphore} will be released when the command is
     * completed.
     * 
     * @param command
     * @param lock
     */
    protected void sendCommand(final WorkerCommand command, final Semaphore lock) {
        commandPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    WorkerHelper.sendCommand(command, lock);
                } catch (TException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
}
